package cn.itcast.function;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.IndexBean;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class SecIndexWindowFunction implements WindowFunction<CleanBean, IndexBean,String, TimeWindow> {

    @Override
    public void apply(String s, TimeWindow timeWindow, Iterable<CleanBean> iterable, Collector<IndexBean> collector) throws Exception {
        CleanBean cleanBean = null;
        for (CleanBean bean : iterable) {
            if(cleanBean==null){
                cleanBean=bean;
            }
            if(cleanBean.getEventTime()<bean.getEventTime()){
                cleanBean=bean;
            }
        }
        IndexBean indexBean = new IndexBean();
        indexBean.setEventTime(cleanBean.getEventTime());
        indexBean.setIndexCode(cleanBean.getSecCode());
        indexBean.setIndexName(cleanBean.getSecName());
        indexBean.setPreClosePrice(cleanBean.getPreClosePrice());
        indexBean.setOpenPrice(cleanBean.getOpenPrice());
        indexBean.setHighPrice(cleanBean.getMaxPrice());
        indexBean.setLowPrice(cleanBean.getMinPrice());
        indexBean.setClosePrice(cleanBean.getTradePrice());
        indexBean.setTradeVol(0L);
        indexBean.setTradeAmt(0L);
        indexBean.setTradeVolDay(cleanBean.getTradeVolumn());
        indexBean.setTradeAmtDay(cleanBean.getTradeAmt());
        indexBean.setTradeTime(DateUtil.longTimeTransfer(cleanBean.getEventTime(), Constant.format_YYYYMMDDHHMMSS));
        indexBean.setSource(cleanBean.getSource());

        collector.collect(indexBean);

    }
}
